import pandas as pd
from typing import Dict, List, Optional, Union
import numpy as np
import akshare as ak
import logging

# 配置日志
logger = logging.getLogger(__name__)

class NorthboundData:
    """北向资金数据类
    处理北向资金流向和持股数据
    """
    
    def __init__(self):
        """初始化北向资金数据类"""
        pass
        
    def get_northbound_flow(self, start_date: str, end_date: str) -> Optional[Dict]:
        """获取北向资金流入流出数据
        
        Args:
            start_date: 开始日期，格式YYYYMMDD
            end_date: 结束日期，格式YYYYMMDD
            
        Returns:
            包含北向资金流入流出数据的字典，包括以下字段：
            - trade_date: 交易日期列表
            - north_money: 北向资金流入金额
            - sh_money: 沪股通资金流入金额
            - sz_money: 深股通资金流入金额
        """
        try:
            logger.info("开始获取北向资金流入流出数据")
            logger.debug(f"请求参数: start_date={start_date}, end_date={end_date}")
            
            # 调用AKShare接口获取数据 - 使用东方财富网数据源
            df = ak.stock_hsgt_north_net_flow_in_em()
            
            if df.empty:
                logger.warning("未获取到北向资金流入流出数据")
                return None
                
            # 转换列名以匹配标准格式
            df = df.rename(columns={
                '日期': 'trade_date',
                '北向资金': 'north_money',
                '沪股通': 'sh_money',
                '深股通': 'sz_money'
            })
            
            # 确保日期格式统一
            df['trade_date'] = pd.to_datetime(df['trade_date']).dt.strftime('%Y%m%d')
            
            # 筛选日期范围内的数据
            df = df[(df['trade_date'] >= start_date) & (df['trade_date'] <= end_date)]
            
            # 按日期升序排序
            df = df.sort_values('trade_date')
            
            logger.info(f"获取到 {len(df)} 条北向资金流入流出数据记录")
            
            return {
                'trade_date': df['trade_date'].values,
                'north_money': df['north_money'].values,
                'sh_money': df['sh_money'].values,
                'sz_money': df['sz_money'].values
            }
            
        except Exception as e:
            logger.error(f"获取北向资金流入流出数据失败: {type(e).__name__} - {str(e)}")
            return None
    
    def get_northbound_acc_flow(self, start_date: str, end_date: str) -> Optional[Dict]:
        """获取北向资金累计流入数据
        
        Args:
            start_date: 开始日期，格式YYYYMMDD
            end_date: 结束日期，格式YYYYMMDD
            
        Returns:
            包含北向资金累计流入数据的字典，包括以下字段：
            - trade_date: 交易日期列表
            - north_money_acc: 北向资金累计流入金额
            - sh_money_acc: 沪股通累计流入金额
            - sz_money_acc: 深股通累计流入金额
        """
        try:
            logger.info("开始获取北向资金累计流入数据")
            logger.debug(f"请求参数: start_date={start_date}, end_date={end_date}")
            
            # 调用AKShare接口获取北向资金累计流入数据
            df = ak.stock_em_hsgt_north_acc_flow_in()
            
            if df.empty:
                logger.warning("未获取到北向资金累计流入数据")
                return None
                
            # 转换列名以匹配标准格式
            df = df.rename(columns={
                '日期': 'trade_date',
                '北向资金累计净流入': 'north_money_acc',
                '沪股通累计净流入': 'sh_money_acc',
                '深股通累计净流入': 'sz_money_acc'
            })
            
            # 确保日期格式统一
            df['trade_date'] = pd.to_datetime(df['trade_date']).dt.strftime('%Y%m%d')
            
            # 筛选日期范围内的数据
            df = df[(df['trade_date'] >= start_date) & (df['trade_date'] <= end_date)]
            
            # 按日期升序排序
            df = df.sort_values('trade_date')
            
            logger.info(f"获取到 {len(df)} 条北向资金累计流入数据记录")
            
            return {
                'trade_date': df['trade_date'].values,
                'north_money_acc': df['north_money_acc'].values,
                'sh_money_acc': df['sh_money_acc'].values,
                'sz_money_acc': df['sz_money_acc'].values
            }
            
        except Exception as e:
            logger.error(f"获取北向资金累计流入数据失败: {type(e).__name__} - {str(e)}")
            return None
    
    def get_northbound_holding(self, start_date: str, end_date: str) -> Optional[Dict]:
        """获取北向资金持股数据
        
        Args:
            start_date: 开始日期，格式YYYYMMDD
            end_date: 结束日期，格式YYYYMMDD
            
        Returns:
            包含北向资金持股数据的字典，包括以下字段：
            - trade_date: 交易日期列表
            - stock_code: 股票代码列表
            - stock_name: 股票名称列表
            - holding_amount: 持股数量列表
            - holding_value: 持股市值列表
            - holding_ratio: 持股占比列表
            - value_change: 持股市值变化列表
            - amount_change: 持股数量变化列表
        """
        try:
            logger.info("开始获取北向资金持股数据")
            logger.debug(f"请求参数: start_date={start_date}, end_date={end_date}")
            
            # 调用AKShare接口获取数据
            df = ak.stock_hsgt_north_holdings()
            
            if df.empty:
                logger.warning("未获取到北向资金持股数据")
                return None
                
            # 转换列名以匹配标准格式
            df = df.rename(columns={
                '日期': 'trade_date',
                '股票代码': 'stock_code',
                '股票名称': 'stock_name',
                '持股数量': 'holding_amount',
                '持股市值': 'holding_value',
                '持股占比': 'holding_ratio',
                '持股市值变化': 'value_change',
                '持股数量变化': 'amount_change'
            })
            
            # 确保日期格式统一
            df['trade_date'] = pd.to_datetime(df['trade_date']).dt.strftime('%Y%m%d')
            
            # 筛选日期范围内的数据
            df = df[(df['trade_date'] >= start_date) & (df['trade_date'] <= end_date)]
            
            # 按日期升序排序
            df = df.sort_values('trade_date')
            
            logger.info(f"获取到 {len(df)} 条北向资金持股数据记录")
            
            return {
                'trade_date': df['trade_date'].values,
                'stock_code': df['stock_code'].values,
                'stock_name': df['stock_name'].values,
                'holding_amount': df['holding_amount'].values,
                'holding_value': df['holding_value'].values,
                'holding_ratio': df['holding_ratio'].values,
                'value_change': df['value_change'].values,
                'amount_change': df['amount_change'].values
            }
            
        except Exception as e:
            logger.error(f"获取北向资金持股数据失败: {type(e).__name__} - {str(e)}")
            return None
    
    def get_northbound_holding_em(self, market: str = "北向") -> Optional[Dict]:
        """获取北向资金持股数据（东方财富版）
        
        Args:
            market: 市场类型，可选值："北向"、"沪股通"、"深股通"
            
        Returns:
            包含北向资金持股数据的字典，包括以下字段：
            - stock_code: 股票代码列表
            - stock_name: 股票名称列表
            - holding_amount: 持股数量列表
            - holding_ratio: 持股占比列表
            - market_value: 持股市值列表
        """
        try:
            logger.info(f"开始获取{market}持股数据")
            
            # 调用AKShare接口获取数据
            df = ak.stock_hsgt_hold_stock_em(market=market)
            
            if df.empty:
                logger.warning(f"未获取到{market}持股数据")
                return None
                
            # 转换列名以匹配标准格式
            df = df.rename(columns={
                '代码': 'stock_code',
                '名称': 'stock_name',
                '持股数量': 'holding_amount',
                '持股市值': 'market_value',
                '持股占比': 'holding_ratio'
            })
            
            logger.info(f"获取到 {len(df)} 条{market}持股数据记录")
            
            return {
                'stock_code': df['stock_code'].values,
                'stock_name': df['stock_name'].values,
                'holding_amount': df['holding_amount'].values,
                'market_value': df['market_value'].values,
                'holding_ratio': df['holding_ratio'].values
            }
            
        except Exception as e:
            logger.error(f"获取{market}持股数据失败: {type(e).__name__} - {str(e)}")
            return None
    
    def get_northbound_stock_ranking(self, market: str = "北向", indicator: str = "今日排行") -> Optional[Dict]:
        """获取北向资金持股个股排行数据
        
        Args:
            market: 市场类型，可选值："北向"、"沪股通"、"深股通"
            indicator: 排行指标，可选值："今日排行"、"3日排行"、"5日排行"、"10日排行"、"月排行"、"季排行"、"年排行"
            
        Returns:
            包含北向资金持股个股排行数据的字典，包括以下字段：
            - stock_code: 股票代码列表
            - stock_name: 股票名称列表
            - close_price: 收盘价列表
            - change_percent: 涨跌幅列表（%）
            - holding_amount: 持股数量列表（万股）
            - holding_value: 持股市值列表（万元）
            - holding_ratio_float: 持股占流通股比列表（%）
            - holding_ratio_total: 持股占总股本比列表（%）
            - increase_amount: 增持股数列表（万股）
            - increase_value: 增持市值列表（万元）
            - increase_value_percent: 增持市值增幅列表（%）
            - increase_ratio_float: 增持占流通股比列表（‰）
            - increase_ratio_total: 增持占总股本比列表（‰）
            - sector: 所属板块列表
            - trade_date: 交易日期列表
        """
        try:
            logger.info(f"开始获取{market}{indicator}数据")
            
            # 调用AKShare接口获取数据
            df = ak.stock_hsgt_hold_stock_em(market=market, indicator=indicator)
            
            if df.empty:
                logger.warning(f"未获取到{market}{indicator}数据")
                return None
            
            # 获取列名，用于动态处理不同indicator下的列名变化
            columns = df.columns.tolist()
            logger.debug(f"原始数据列名: {columns}")
            
            # 基础列名映射
            column_map = {
                '代码': 'stock_code',
                '名称': 'stock_name',
                '今日收盘价': 'close_price',
                '今日涨跌幅': 'change_percent',
                '所属板块': 'sector',
                '日期': 'trade_date'
            }
            
            # 根据indicator动态添加持股和增持相关的列名映射
            # 持股相关列名
            holding_columns = [col for col in columns if '持股' in col]
            if len(holding_columns) >= 4:
                column_map.update({
                    holding_columns[0]: 'holding_amount',  # 持股数量
                    holding_columns[1]: 'holding_value',  # 持股市值
                    holding_columns[2]: 'holding_ratio_float',  # 持股占流通股比
                    holding_columns[3]: 'holding_ratio_total'   # 持股占总股本比
                })
            
            # 增持相关列名
            increase_columns = [col for col in columns if '增持' in col]
            if len(increase_columns) >= 5:
                column_map.update({
                    increase_columns[0]: 'increase_amount',  # 增持股数
                    increase_columns[1]: 'increase_value',  # 增持市值
                    increase_columns[2]: 'increase_value_percent',  # 增持市值增幅
                    increase_columns[3]: 'increase_ratio_float',  # 增持占流通股比
                    increase_columns[4]: 'increase_ratio_total'   # 增持占总股本比
                })
            
            # 转换列名
            df_renamed = df.rename(columns=column_map)
            logger.debug(f"转换后的列名: {df_renamed.columns.tolist()}")
            
            # 构建结果字典
            result = {}
            for col in df_renamed.columns:
                result[col] = df_renamed[col].values
            
            logger.info(f"获取到 {len(df)} 条{market}{indicator}数据记录")
            
            return result
            
        except Exception as e:
            logger.error(f"获取{market}{indicator}数据失败: {type(e).__name__} - {str(e)}")
            return None